package org.example.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * NIO多路复用器
 */

public class NIOServer {

    private  int port;
    private Selector selector;

    private ThreadPoolExecutor poolExecutor ;

    public static void main(String[] args) {
        NIOServer server = new NIOServer(5000);
        server.start();
    }


    public NIOServer(int port){
        this.port = port;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(30);
        poolExecutor = new ThreadPoolExecutor(10,25,30, TimeUnit.SECONDS,workQueue);
    }


    private void init(){
        ServerSocketChannel ssc = null;
        try {
            ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);//设置非阻塞
            ssc.bind(new InetSocketAddress(port));
            this.selector = Selector.open();//打开选择器 对于内核来说就是epoll_create
            ssc.register(selector, SelectionKey.OP_ACCEPT);//将选择器注册，内核为epoll_ctrl
            System.out.println("NIO server is staring in "+port+"...");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 接受链接，注册到selector中
     * @param key
     */
    public void accept(SelectionKey key) {
        try {
            ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
            SocketChannel sc = ssc.accept();
            sc.configureBlocking(false);
            sc.register(selector,SelectionKey.OP_READ);
            System.out.println("accept a client : " + sc.socket().getInetAddress().getHostName());
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public void start(){
        this.init();
        try {
        while(true){
            /**
             * select为阻塞，相当于内核的epoll_wait
             */
            int events = this.selector.select(500);
            if(events > 0 ){
                //获取注册到Selector上的key
                Iterator<SelectionKey> tor = selector.selectedKeys().iterator();
                while(tor.hasNext()){
                    SelectionKey selectionKey = tor.next();
                    tor.remove();
                    // 判断是哪个事件
                    if(selectionKey.isAcceptable()){
                        //链接请求
                        accept(selectionKey);
                        System.out.println(selectionKey.attachment() + " - 接受请求事件");
                    }else{
                        poolExecutor.submit(()->{
                            executeProcess(selectionKey);
                        });
                    }

                }
            }

         }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void executeProcess(SelectionKey selectionKey ){
        try {
        if(selectionKey.isReadable()){
            // 读数据
            SocketChannel sc = (SocketChannel) selectionKey.channel();
            ByteBuffer buff = ByteBuffer.allocate(1024);
            sc.read(buff);
            buff.flip();
            System.out.println("收到客户端"+sc.socket().getInetAddress().getHostName()+"的数据："+new String(buff.array()));
            buff.clear();
            ByteBuffer out = ByteBuffer.allocate(256);
            out.put("服务端已收到消息".getBytes("utf-8"));
            out.flip();
            sc.write(out);
            selectionKey.cancel();
        }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }




}
